package global;

import bufmgr.HashEntryNotFoundException;
import bufmgr.InvalidFrameNumberException;
import bufmgr.PageUnpinnedException;
import bufmgr.ReplacerException;
import heap.*;
import iterator.CondExpr;
import iterator.FileScan;
import iterator.FldSpec;
import iterator.RelSpec;

import java.io.*;
import java.util.ArrayList;
import java.util.List;

import btree.*;
import diskmgr.PCounter;
import va.BitSet;
import va.Range;
import va.VAEntry;
import va.VAIndexAdapter;
import va_file.VAFile;

public class BatchDelete {

    // Takes in a file and cycles through line by line, deleting the tuples
    // based on the condition presented in each line

	public static void Delete(String fileName, String relName) throws HFException, HFBufMgrException, HFDiskMgrException, GetFileEntryException, ConstructPageException, AddFileEntryException, IOException, PageUnpinnedException, InvalidFrameNumberException, HashEntryNotFoundException, ReplacerException {
		// initialize counter so we can count pages written and read
		PCounter.initialize();
	
		// get information for related indexes so they can also be updated
		BatchIndex indexer = new BatchIndex(relName, false);
		
		// read the deletion criteria from the file
		BufferedReader br = new BufferedReader(new FileReader(fileName));
	    int attrNum = 0;
	    attrNum = Integer.parseInt(br.readLine().trim());
	    String line = br.readLine().trim();
	    Schema ts;
	    if ((ts = Batch.getAttribTypes(line, attrNum)) == null)
	    {
	    	System.err.println("Error reading file.");
	    	return;
	    }
	    
		while ( (line = br.readLine()) != null )
		{
			// delete for the condition on the current line in the file
			deleteLine(line, relName, ts);
		}

        indexer.close();
	}

    // Performs a delete for a specific line. First deletes from the indexes, then delete from the
    // data set

    private static void deleteLine(String line, String relName, Schema schema) throws HFDiskMgrException, HFException, IOException, ConstructPageException, AddFileEntryException, GetFileEntryException, HFBufMgrException, PageUnpinnedException, InvalidFrameNumberException, HashEntryNotFoundException, ReplacerException {
        FileScan fs = null;
        VAIndexAdapter[] adapters = null;
        CondExpr[] expr = getCondition(line, schema);
        FldSpec[] projlist = CommandLine.getFldSpecForAll(schema.attrType);
        List<RID> rids = new ArrayList<RID>();

        // First, delete the indexes and gather the RIDs to delete from the data set
        try
        {
            IDXSchema[] indexSchemas = IDXSchema.GetIndexes(relName);
            adapters = new VAIndexAdapter[indexSchemas.length];

            // Initialize all the adapters
            for (int i = 0; i < indexSchemas.length; i++) {
                adapters[i] = new VAIndexAdapter(indexSchemas[i], false);
            }

            // scan file to get list of RIDs to be deleted.
            fs = new FileScan(relName, schema.attrType, schema.attrSize, (short)schema.attrType.length, projlist.length, projlist, expr);
            Tuple currentTuple = fs.get_next();
            while (currentTuple != null)
            {
                RID currentRID = fs.scan.currentRID;

                // Cycle through the indexes and remove the relevant rows
                for (int i = 0; i < adapters.length; i++) {
                    VAIndexAdapter adapter = adapters[i];
                    IDXSchema indexSchema = adapter.Index;

                    // Create the VAEntry based on the relevant tuple column
                    Vector100Dtype vector = currentTuple.get100DVectFld(indexSchema.ColumnID);
                    Range[] ranges = VAFile.GetDefaultRanges(indexSchema.Bits);
                    BitSet approximation = VAEntry.CreateApproximation(vector, ranges, indexSchema.Bits);
                    VAEntry entry = new VAEntry(approximation, currentRID);

                    // Remove the entry from the index file
                    adapter.deleteEntry(entry);
                }

                // Add rid to the list to be removed
                rids.add(currentRID);

                currentTuple = fs.get_next();
            }
        }
        catch (Exception e)
        {
            System.err.println(e.getMessage());
        }
        finally
        {
            if (fs!=null) fs.close();

            if (adapters != null) {
                // Close all the adapters
                for (int i = 0; i < adapters.length; i++) {
                    adapters[i].close();
                }
            }
        }

        // Now go through the data set and remove the relevant rids
        // Note: this is kept separate from the above loop because I'm afraid of buffer management issues with having the heapfile and
        // file scanner open at the same time
        try
        {
            Heapfile hf = new Heapfile(relName);
            for (int i=0; i<rids.size(); i++)
            {
                try
                {
                    hf.deleteRecord(rids.get(i));
                }
                catch (Exception e)
                {
                    System.err.println("Error deleting tuple for condition: " + line);
                }
            }
            System.out.println("Deleted " + rids.size() + " tuples from the data set.");
        }
        catch (Exception e)
        {
            System.err.println("Could not open heap file for deletion.");
        }
    }

    // Formats the condition to query on for a line

	private static CondExpr[] getCondition(String line, Schema ts)
	{
		String[] conditions = line.split("\\s+");
		if (conditions.length < 2)
		{
			System.err.println("Invalid deletion condition; wrong number of entries on line: " + line);
			return null;
		}
		int att_num;
		try
		{
			att_num = Integer.parseInt(conditions[0]);
		}
		catch (Exception e)
		{
			System.err.println("Invalid deletion condition. Expected an integer, found: " + conditions[0]);
			return null;
		}
		
		CondExpr[] expr = new CondExpr[2];
		expr[0] = new CondExpr();
		expr[0].next  = null;
		expr[0].op    = new AttrOperator(AttrOperator.aopEQ);
		expr[0].type1 = new AttrType(AttrType.attrSymbol);
		expr[0].type2 = ts.attrType[att_num-1];
		expr[0].operand1.symbol = new FldSpec (new RelSpec(RelSpec.outer),att_num);
		expr[1] = null;
		
		switch (ts.attrType[att_num-1].attrType)
		{
		case AttrType.attrInteger:
			int intVal = Integer.parseInt(conditions[1]);
			expr[0].operand2.integer = intVal;
			break;
		case AttrType.attrReal:
			Float floatVal = Float.parseFloat(conditions[1]);
			expr[0].operand2.real = floatVal;
			break;
		case AttrType.attrString:
			// reconstruct string (may have been split on spaces)
			String value = "";
			for (int i=1; i<conditions.length; i++)
			{
				if (!value.isEmpty())
				{
					value += " ";
				}
				value += conditions[i];
			}
			expr[0].operand2.string = value;
			break;
		case AttrType.attrVector100D:
			if (conditions.length < 101)
			{
				System.err.println("Invalid deletion condition for vector column. Expected 100 integers.");
				return null;
			}
			String[] input = new String[100];
			System.arraycopy(conditions, 1, input, 0, 100);
			Vector100Dtype vector = new Vector100Dtype(input);
			expr[0].operand2.vector = vector;
			expr[0].distance = 0;
			break;
		}
		
		return expr;
	}

}
